/**
 * Copyright 2013-2015
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.terramagnet.learn.rabbitmq.producer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import javax.annotation.Resource;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.stereotype.Service;
import org.terramagnet.learn.rabbitmq.producer.util.Hex62Util;

/**
 *
 * @author lip
 */
@Service
public class Producer {

    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final ObjectMapper jsonUtil = new ObjectMapper();

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private JdbcTemplate jdbcTemplate;
    /**
     * 生产数据来源. 表名或{@code null}
     */
    private String from="message1";

    public List<MessageSample> produce(int index, int size) {
        try {
            if (StringUtils.isNotBlank(from)) {
                return jdbcTemplate.query(String.format("select id,message from %s limit ?,?", from), new RowMapper<MessageSample>() {

                    @Override
                    public MessageSample mapRow(ResultSet rs, int rowNum) throws SQLException {
                        MessageSample entity=new MessageSample();
                        entity.setId(rs.getInt("id"));
                        entity.setMessage(rs.getString("message"));
                        return entity;
                    }
                }, index, size);
            }
        } catch (DataAccessException ex) {
            logger.debug(String.format("数据来源表【%s】不存在", from), ex);
            from=null;
            logger.info("切换到uuid模式");
        }

        List<MessageSample> messages = new ArrayList<MessageSample>();
        while (messages.size() < size) {
            messages.add(new MessageSample(Hex62Util.generateShortUuid()));
        }
        return messages;
    }

    private List<MessageSample> buffer;
    private int i;
    public void run() {
        i = 0;
        buffer=null;
        while (hasNext()) {
            int size = (int) (Math.random() * 100) + 10000;
            buffer = produce(i, size);
            i += buffer.size();
            logger.info("Prepare to send {} messages...",buffer.size());
            for (MessageSample p : buffer) {
                try {
                    rabbitTemplate.convertAndSend(jsonUtil.writeValueAsString(p));
                } catch (JsonProcessingException ex) {
                    logger.debug("json 序列化失败", ex);
                }
            }
            logger.info("Send complete!");
        }
    }
    
    public boolean hasNext(){
        if(StringUtils.isBlank(from)){
            return i<500000;
        }else{
            return buffer==null || !buffer.isEmpty();
        }
    }
    
}
